package app.dws;

import app.BaseAppV1;
import beans.ExamQuestionBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import utils.AtguiguUtil;
import utils.FlinkSinkUtil;

import java.text.SimpleDateFormat;
import java.time.Duration;

public class Dws_15_ExamQuestionCountWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_15_ExamQuestionCountWindow().init(
                5013,
                2,
                "Dws_153_ExamQuestionCountWindow",
                Constant.DWD_EXAM_QUESTION_DETAIL
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 对数据进行etl 清洗
        DataStreamSource<String> etledStream = etl(stream);
        //将数据转换成pojo
        SingleOutputStreamOperator<ExamQuestionBean> beanStream = parseToPOJO(etledStream);
        //对数据流进行分组开窗聚合
        SingleOutputStreamOperator<ExamQuestionBean> windowAndAggedStream = windowAndAgg(beanStream);
        windowAndAggedStream.print();
        //将数据写入ck
        writeToCK(windowAndAggedStream);
    }

    private void writeToCK(SingleOutputStreamOperator<ExamQuestionBean> windowAndAggedStream) {
        windowAndAggedStream.addSink(FlinkSinkUtil.getClickHouseSink(Constant.DWS_EXAM_QUESTION_SCORE_WINDOW,ExamQuestionBean.class));
    }

    private SingleOutputStreamOperator<ExamQuestionBean> windowAndAgg(SingleOutputStreamOperator<ExamQuestionBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ExamQuestionBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner( (examQuestionBean, l) -> examQuestionBean.getTs())
                )
                .keyBy(examQuestionBean -> examQuestionBean.getQuestionId())
                .map(new RichMapFunction<ExamQuestionBean, ExamQuestionBean>() {

                    private ValueState<String> state;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        state = getRuntimeContext().getState(new ValueStateDescriptor<String>("state", String.class));
                    }

                    @Override
                    public ExamQuestionBean map(ExamQuestionBean examQuestionBean) throws Exception {
                        //获取状态 如果没有则是该题目下第一次做题
                        String value = state.value();
                        if (value == null){
                            state.update(examQuestionBean.getUserId());
                            if ("1".equals(examQuestionBean.getIsCorrect()) ){
                                examQuestionBean.setAruvCt(examQuestionBean.getAruvCt() + 1);
                                examQuestionBean.setUvCt(examQuestionBean.getUvCt() + 1);
                            }
                        }else if (value.equals(examQuestionBean.getUserId())){  //不是第一次答题
                            //答题次数+1
                            examQuestionBean.setAnsCount(examQuestionBean.getAnsCount() + 1);
                        }
                        return examQuestionBean;
                    }
                })
                .keyBy(new KeySelector<ExamQuestionBean, Tuple2<String,String>>() {
                    @Override
                    public Tuple2<String, String> getKey(ExamQuestionBean examQuestionBean) throws Exception {
                        return new Tuple2<>(examQuestionBean.getQuestionId(),examQuestionBean.getUserId());
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<ExamQuestionBean>() {
                            @Override
                            public ExamQuestionBean reduce(ExamQuestionBean b1, ExamQuestionBean b2) throws Exception {
                                b1.setAnsCount(b1.getAnsCount() + b2.getAnsCount());
                                b1.setAnsrightCount(b1.getAnsrightCount() + b2.getAnsrightCount());
                                return b1;
                            }
                        }, new ProcessWindowFunction<ExamQuestionBean,ExamQuestionBean, Tuple2<String, String>, TimeWindow>() {
                            @Override
                            public void process(Tuple2<String, String> stringStringTuple2, Context context, Iterable<ExamQuestionBean> iterable, Collector<ExamQuestionBean> collector) throws Exception {
                                ExamQuestionBean bean = iterable.iterator().next();
                                bean.setStt(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(context.window().getStart()));
                                bean.setEdt(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(context.window().getEnd()));
                                bean.setTs(System.currentTimeMillis());
                                collector.collect(bean);
                            }
                        }

                );
    }

    private SingleOutputStreamOperator<ExamQuestionBean> parseToPOJO(DataStreamSource<String> etledStream) {
        return etledStream
                .map(new
                             MapFunction<String, ExamQuestionBean>() {
                                 @Override
                                 public ExamQuestionBean map(String s) throws Exception {
                                     JSONObject data = JSON.parseObject(s);
                                     String questionId = data.getString("question_id");
                                     String isCorrect = data.getString("is_correct");
                                     String userId = data.getString("user_id");
                                     Long ansCount = 1L;
                                     Long ansRightCount = 0L;
                                     Long arUvCt = 0L;
                                     if ("1".equals(data.getString("is_correct"))){
                                         ansRightCount = 1L;
                                     }
                                     Long UvCt = 1L;
                                     return new ExamQuestionBean("","",questionId,userId,isCorrect,ansCount,ansRightCount,arUvCt, UvCt, AtguiguUtil.dateToTs(data.getString("create_time")));
                                 }
                             });

    }

    private DataStreamSource<String> etl(DataStreamSource<String> stream) {
        //暂时不需要清洗
        return stream;
    }
}